package com.zws.cucumber.util.cass.backdoor;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.LineEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;

import static com.zws.cucumber.util.cass.backdoor.ZwsCassBackdoorOp.*;

public class ZwsCassBackdoorClientImpl implements ZwsCassBackdoorClient {

    private static final Logger logger = LoggerFactory.getLogger(ZwsCassBackdoorClientImpl.class);

    private static EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);

    private Channel channel;

    private AtomicReference<Promise<String>> responsePromiseRef = new AtomicReference<>(null);

    public ZwsCassBackdoorClientImpl(int port) {
        try {
            long s = System.nanoTime();
            channel = new Bootstrap()
                    .group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE));
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new LineEncoder());
                            pipeline.addLast(new ChannelDuplexHandler() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    logger.trace("backdoor client recv: {}", msg);
                                    if (responsePromiseRef.get() != null) {
                                        responsePromiseRef.get().setSuccess((String) msg);
                                    }
                                    super.channelRead(ctx, msg);
                                }

                                @Override
                                public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                    logger.trace("backdoor client send in handler: {}", msg);
                                    super.write(ctx, msg, promise);
                                }
                            });
                        }
                    })
                    .connect("127.0.0.1", port)
                    .sync()
                    .channel();
            logger.info("backdoor client connected, elapsed: {}ms", (System.nanoTime() - s) / 1000000);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    // String keyspace, String ... tables
    public BackdoorResp sendOperationRequest(ZwsCassBackdoorOp operation, String argumentLine) {
        long s = System.nanoTime();
        if (responsePromiseRef.compareAndSet(null, new DefaultPromise<>(eventLoopGroup.next()))) {
            logger.trace("backdoor client send, operation: {}, argumentLine: {}", operation, argumentLine);

        } else {
            throw new IllegalStateException();
        }
        try {
            channel.writeAndFlush(operation + (Objects.nonNull(argumentLine) ? " " + argumentLine : ""));
            String resp = responsePromiseRef.get().get();
            if (resp.startsWith("Ok.")) {
                logger.debug("backdoor operation [{}:{}] succeed, elapsed: {}ms, resp: {}", operation, argumentLine, (System.nanoTime() - s) / 1000000, resp);
                return new BackdoorResp(true, resp.substring(3));
            } else {
                logger.error("backdoor operation failed, elapsed: {}ms\n  operation: {}\n  arguments: {}\n  resp: {}",
                        (System.nanoTime() - s) / 1000000,
                        operation,
                        argumentLine,
                        resp);
                return new BackdoorResp(false, resp);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            responsePromiseRef.set(null);
        }
        return new BackdoorResp(false, "");
    }

    @Override
    public List<TableHash> listTableHashes(String keyspace) {
        return List.of();
    }

    @Override
    public void updateTableHashes(String keyspace, List<TableHash> tableHashes) {
        String tableHashesStr = tableHashes
                .stream()
                .map(th -> th.table() + " " + th.hash())
                .reduce((a, b) -> a + " " + b)
                .orElse("");
        sendOperationRequest(TABLE_HASH_UPDATE, keyspace + " " + tableHashesStr);
    }

    @Override
    public void flushTableHashes(String keyspace) {
        sendOperationRequest(TABLE_HASH_FLUSH, keyspace);
    }

    public void truncate(String keyspace, String... tables) {
        if (tables == null || tables.length == 0) {
            logger.info("backdoor truncate succeed, shortcut");
            return;
        }
        sendOperationRequest(TABLE_TRUNCATE, keyspace + " " + String.join(" ", tables));
    }

    public void takeSnapshot(String snapshotName, String keyspace) {
        sendOperationRequest(SNAPSHOT_TAKE, snapshotName + " " + keyspace);
    }

    public void restoreSnapshot(String snapshotName, String keyspace) {
        sendOperationRequest(SNAPSHOT_RESTORE, snapshotName + " " + keyspace);
    }

    public void deleteSnapshot(String snapshotName) {
        sendOperationRequest(SNAPSHOT_DELETE, snapshotName);
    }

    public void clearSnapshots() {
        sendOperationRequest(SNAPSHOT_CLEAR, null);
    }

    public void enableTimeout(String dataManipulationType, String targetTable) {
        if (dataManipulationType == null) {
            dataManipulationType = "NO_DATA_MANIPULATION_TYPE";
        }
        dataManipulationType = dataManipulationType.toUpperCase();

        if (targetTable == null) {
            targetTable = "NO_TARGET_TABLE";
        }
        sendOperationRequest(TIMEOUT_ENABLE, dataManipulationType + " " + targetTable);
    }

    public void disableTimeout() {
        sendOperationRequest(TIMEOUT_DISABLE, null);
    }

    record BackdoorResp(boolean isSuccess, String resp) {}

}
